package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.json.JSONUtil;
import com.hmdp.dto.Result;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.UserHolder;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.aop.framework.AopContext;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Resource
    private ISeckillVoucherService seckillVoucherService;

    @Resource
    private RedisIdWorker redisIdWorker;

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private RedissonClient redissonClient;

    @Resource
    RabbitTemplate rabbitTemplate;

    // 使用redis的Stream作为阻塞队列
//    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
//    static {
//        SECKILL_SCRIPT = new DefaultRedisScript<>();
//        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
//        SECKILL_SCRIPT.setResultType(Long.class);
//    }
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckillRabbit.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    private IVoucherOrderService proxy;

//    @PostConstruct
//    private void init(){
//        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
//    }

//    private class VoucherOrderHandler implements Runnable {
//        String queueName = "stream.orders";
//        @Override
//        public void run() {
//            while(true){
//                try {
//                    // 1、从消息队列中获取信息XREADGROUP GROUP g1 c1 Count 1 BLOCK 2000 STREAMS stream.orders >
//                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
//                            Consumer.from("g1", "c1"),
//                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
//                            StreamOffset.create(queueName, ReadOffset.lastConsumed())
//                    );
//                    // 2、判断是否有消息，没有则再次获取
//                    if(list == null || list.isEmpty()){
//                        continue;
//                    }
//                    // 3、解析订单信息，用于存储到数据库中
//                    MapRecord<String, Object, Object> record = list.get(0);
//                    Map<Object, Object> value = record.getValue();
//                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
//                    // 4、执行方法存入数据库
//                    handleVoucherOrder(voucherOrder);
//                    // 5、XACK消息
//                    stringRedisTemplate.opsForStream().acknowledge(queueName,"g1", record.getId());
//                } catch (Exception e) {
//                    log.error("处理订单异常" + e);
//                    handlePendingList();
//                }
//            }
//        }
// 阻塞队列的逻辑：
// 第一版（使用JVM的阻塞队列）：请求打到seckillVoucher上，然后上面判断是否有资格 -> 放入阻塞BlockingQueue<VoucherOrder> orderTasks中 ->
    // 并在类初始化时就开启新线程不断循环处理订单，在其中调用handleVoucherOrder方法 -> 在handleVoucherOreder中获取分布式锁并调用createVoucherOrder保存订单信息并删减缓存

//  第二版（使用Redis的Stream消息队列）：修改seckill.lua脚本（也就是现在的版本）多了一步判断有资格之后会向消息队列中存储用户id，优惠劵id和订单id ->
//    在本类初始化时就开启新线程不断循环获取Stream消息队列中的消息，在其中调用handleVoucherOrder方法 -> -> 在handleVoucherOreder中获取分布式锁并调用createVoucherOrder保存订单信息并删减缓存

//  第三版（使用RabbitMQ）：请求打到seckillVoucher上，然后上面判断是否有资格，有则将其放入到Rabbit中，并交由RabbitMQConsumer来处理订单 ->
//  Consumer直接调用handleVoucherOrder方法 -> 在handleVoucherOreder中获取分布式锁并调用createVoucherOrder保存订单信息并删减缓存
//

        // Stream阻塞队列的等待队列，获取信息后需要等待其他消费者消费完毕
//        private void handlePendingList() {
//            while(true){
//                try {
//                    // 1、从消息队列中获取信息XREADGROUP GROUP g1 c1 Count 1 BLOCK 2000 STREAMS stream.orders >
//                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
//                            Consumer.from("g1", "c1"),
//                            StreamReadOptions.empty().count(1),
//                            StreamOffset.create(queueName, ReadOffset.from("0"))
//                    );
//                    // 2、判断是否有消息，没有则再次获取
//                    if(list == null || list.isEmpty()){
//                        break;
//                    }
//                    // 3、解析订单信息，用于存储到数据库中
//                    MapRecord<String, Object, Object> record = list.get(0);
//                    Map<Object, Object> value = record.getValue();
//                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
//                    // 4、执行方法存入数据库
//                    handleVoucherOrder(voucherOrder);
//                    // 5、XACK消息
//                    stringRedisTemplate.opsForStream().acknowledge(queueName,"g1", record.getId());
//                } catch (Exception e) {
//                    log.error("处理pending-list异常" + e);
//                }
//            }
//        }
//    }

    // JVM阻塞队列的第一版方法
//    private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);
//    private class VoucherOrderHandler implements Runnable{
//
//        @Override
//        public void run() {
//            while(true){
//                try {
//                    VoucherOrder voucherOrder = orderTasks.take();
//                    handleVoucherOrder(voucherOrder);
//                } catch (Exception e) {
//                    log.error("处理订单异常" + e);
//                }
//            }
//        }
//    }
    // Stream的阻塞队列实现的方法
//    @Override
//    public Result seckillVoucher(Long voucherId) {
//        // 获取用户id
//        Long userId = UserHolder.getUser().getId();
//        long orderId = redisIdWorker.nextId("order");
//        // 执行lua脚本，判断是否有获取资格
//        Long result = stringRedisTemplate.execute(
//                SECKILL_SCRIPT,
//                Collections.emptyList(),
//                voucherId.toString(), userId.toString(), String.valueOf(orderId)
//        );
//        // 判断结果，为1库存不足，2抢过票了，0则通过
//        int r = result.intValue();
//        if(r != 0){
//            return Result.fail(r == 1 ? "库存不足": "不可重复下单");
//        }
//        // 获取代理类对象
//        proxy = (IVoucherOrderService) AopContext.currentProxy();
//        // 返回订单id
//        return Result.ok(orderId);
//    }
    @Override
    public Result seckillVoucher(Long voucherId) {
        // 获取用户id
        Long userId = UserHolder.getUser().getId();
        long orderId = redisIdWorker.nextId("order");
        // 执行lua脚本，判断是否有获取资格
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(), userId.toString()
        );
        // 判断结果，为1库存不足，2抢过票了，0则通过
        int r = result.intValue();
        if(r != 0){
            return Result.fail(r == 1 ? "库存不足": "不可重复下单");
        }
        // 获取代理类对象
        proxy = (IVoucherOrderService) AopContext.currentProxy();

        // 放入阻塞队列中信息下一步的订单处理(放入
        // 封装一下VoucherOrder
        VoucherOrder voucherOrder = new VoucherOrder();
        voucherOrder.setUserId(userId);
        voucherOrder.setVoucherId(voucherId);
        voucherOrder.setId(orderId);
        String message = JSONUtil.toJsonStr(voucherOrder);
        rabbitTemplate.convertAndSend("VoucherOrder.queue", message);
        // 返回订单id
        return Result.ok(orderId);
    }

    // 使用JVM自带的阻塞队列
//    @Override
//    public Result seckillVoucher(Long voucherId) {
//        // 获取用户id
//        Long userId = UserHolder.getUser().getId();
//        // 执行lua脚本，判断是否有获取资格
//        Long result = stringRedisTemplate.execute(
//                SECKILL_SCRIPT,
//                Collections.emptyList(),
//                voucherId.toString(), userId.toString()
//        );
//        // 判断结果，为1库存不足，2抢过票了，0则通过
//        int r = result.intValue();
//        if(r != 0){
//            return Result.fail(r == 1 ? "库存不足": "不可重复下单");
//        }
//        //  创建订单，并将订单信息保存在阻塞队列中，等待写入线程异步的获取并执行
//        VoucherOrder order = new VoucherOrder();
//        long orderId = redisIdWorker.nextId("order");
//        order.setId(orderId);
//        order.setVoucherId(voucherId);
//        order.setUserId(userId);
//        // 加入阻塞队列中
//        orderTasks.add(order);
//
//        // 获取代理类对象
//        proxy = (IVoucherOrderService) AopContext.currentProxy();
//        // 返回订单id
//        return Result.ok(orderId);
//    }

    public void handleVoucherOrder(VoucherOrder voucherOrder) {
        Long userId = voucherOrder.getUserId();
        // 创建锁对象
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        // 获取锁
        boolean isLock = lock.tryLock();
        // 判断是否获取锁成功
        if(!isLock){
            log.error("不允许重复下单");
            return;
        }
        try{
            proxy.createVoucherOrder(voucherOrder);
        } finally {
            lock.unlock();
        }
    }

    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        // 4、一人一单 (经过异步的优化，在前面已经判断过进行了订单的创建放在阻塞队列中，此函数只需要通过voucherOrder的更改即可，但双重保险（其实前面通过redis已经判断过该劵所有订购的用户）
        // 4.1 通过用户id查询订单 （异步为新线程，无法从当前线程取出登录所存的user信息，通过订单取出）
        //Long userId = UserHolder.getUser().getId();
        Long userId = voucherOrder.getUserId();

        // 4.2 存在则返回（二次查询数据库验证，但在seckillRabbit中已经与缓存校验过
        int count = Math.toIntExact(query().eq("user_id", userId).eq("voucher_id", voucherOrder.getVoucherId()).count());
        // 4.3 不存在则进行
        if(count > 0){
            // 扣减失败
            log.error("用户已经抢购过该限时劵");
            return ;
        }
        // 5、时间正常，库存正常则开启订单
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1") // set stock = stock - 1
                .eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0) // where
                .update();
        if(!success){
            log.error("库存不足");
            return ;
        }

        save(voucherOrder);
    }
}
